set spark.sql.crossJoin.enabled=true;

-- 汇总
with union_data as (

select 
    should_bagging_site_code         as site_code
    , should_bagging_site_name       as site_name
    , should_bagging_site_type       as site_type
    , sign_date
    , count(*)                                                                          as should_bagging_cnt
    , sum(case 
            when should_bagging_site_code = nvl(actual_bagging_site_code, '0') then 1 
            else 0 end
      )                                                                                 as actual_bagging_cnt
    , 0                                                                                 as backfill_bagging_cnt
    , 0                                                                                 as unbagging_operate_cnt
from 
    jms_dm.dm_outport_bagging_fine_detail_dt 
where 
    dt = '{{ execution_date | cst_ds }}'
group by 
    should_bagging_site_code
    , should_bagging_site_name
    , should_bagging_site_type
    , sign_date

union all 

select 
    actual_bagging_site_code
    , actual_bagging_site_name
    , actual_bagging_site_type
    , sign_date
    , 0                                                                        as should_bagging_cnt
    , 0                                                                        as actual_bagging_cnt
    , sum(case 
            when should_bagging_site_code <> actual_bagging_site_code 
                 and actual_bagging_site_type <= should_bagging_site_type
                 then 1 
            else 0 
          end 
    )                                                                          as backfill_bagging_cnt
    , 0                                                                        as unbagging_operate_cnt
from 
    jms_dm.dm_outport_bagging_fine_detail_dt 
where 
    dt = '{{ execution_date | cst_ds }}'
group by 
    actual_bagging_site_code
    , actual_bagging_site_name
    , actual_bagging_site_type
    , sign_date

union all 

select 
    end_center_code 
    , end_center_name 
    , '4' -- 类型：中心
    , sign_date
    , 0                                                                        as should_bagging_cnt
    , 0                                                                        as actual_bagging_cnt
    , 0                                                                        as backfill_bagging_cnt
    , sum(case when
                should_bagging_site_code is not null and actual_bagging_site_code is null 
                    then 1 else 0
          end
    )                                                                          as unbagging_operate_cnt
from 
    jms_dm.dm_outport_bagging_fine_detail_dt 
where 
    dt = '{{ execution_date | cst_ds }}'
group by 
    end_center_code 
    , end_center_name 
    , sign_date
)
, center_sum as (
    select                               
        site_code                  -- 站点CODE
        , site_name                -- 站点名称
        , site_type                -- 站点类型
        , sign_date                -- 签收日期
        , agent_code               -- 站点所属代理区CODE
        , agent_name               -- 站点所属代理区名称
        , virt_code                -- 站点所属虚拟代理区CODE
        , virt_name                -- 站点所属虚拟代理区名称
        , should_bagging_cnt       -- 应建包票数
        , actual_bagging_cnt       -- 实建包票数
        , not_bagging_cnt          -- 未建包票数
        , backfill_bagging_cnt     -- 补建包票数
        , unbagging_operate_cnt    -- 未建包小件操作票数
        , (should_bagging_cnt - actual_bagging_cnt) * config.unpack_fines   as not_bagging_fine_amt         -- 未建包罚款标识
        , backfill_bagging_cnt  * config.pack_award                         as backfill_bagging_reward_amt  -- 补建包奖励标识
        , unbagging_operate_cnt * config.staff_award                        as unbagging_operate_reward_amt -- 小件操作补贴
    from(
        select 
            union_data.site_code
            ,union_data.site_name
            ,union_data.site_type
            ,union_data.sign_date
            ,dim_network.agent_code
            ,dim_network.agent_name
            ,dim_network.virt_code
            ,dim_network.virt_name
            ,sum(union_data.should_bagging_cnt)      as should_bagging_cnt
            ,sum(union_data.actual_bagging_cnt)      as actual_bagging_cnt
            ,sum(union_data.should_bagging_cnt) - sum(union_data.actual_bagging_cnt) as not_bagging_cnt
            ,sum(union_data.backfill_bagging_cnt)    as backfill_bagging_cnt
            ,sum(union_data.unbagging_operate_cnt)   as unbagging_operate_cnt
        from 
            union_data
        left join 
            jms_dim.dim_network_whole_massage dim_network 
                on union_data.site_code = dim_network.code
        group by 
             union_data.site_code
            ,union_data.site_name
            ,union_data.site_type
            ,dim_network.agent_code
            ,dim_network.agent_name
            ,dim_network.virt_code
            ,dim_network.virt_name
            ,union_data.sign_date
        having sum(union_data.should_bagging_cnt) + sum(union_data.actual_bagging_cnt) + sum(union_data.backfill_bagging_cnt) + sum(union_data.unbagging_operate_cnt) > 0
    ) a
    left join
        jms_dim.dim_tab_pack_money_config_dt config
            on config.dt = '{{ execution_date | cst_ds }}'
           and config.status = 0
) 
insert overwrite table jms_dm.dm_outport_bagging_fine_sum_dt partition (dt = '{{ execution_date | cst_ds }}')
select * from center_sum
union all 
select 
    agent_code   as site_code               
    , agent_name as site_name             
    , '2'        as site_type             
    , sign_date             
    , agent_code            
    , agent_name            
    , virt_code             
    , virt_name             
    , sum(should_bagging_cnt)     as should_bagging_cnt   
    , sum(actual_bagging_cnt)     as actual_bagging_cnt   
    , sum(not_bagging_cnt)        as not_bagging_cnt      
    , sum(backfill_bagging_cnt)   as backfill_bagging_cnt 
    , sum(unbagging_operate_cnt)  as unbagging_operate_cnt
    , sum(not_bagging_fine_amt)          as not_bagging_fine_amt
    , sum(backfill_bagging_reward_amt)   as backfill_bagging_reward_amt
    , sum(unbagging_operate_reward_amt)  as unbagging_operate_reward_amt
from
    center_sum
group by 
      agent_code
    , agent_name
    , virt_code
    , virt_name
    , sign_date
-- cross join(
--         select * 
--             , row_number() over(order by UPDATE_TIME desc) as rn 
--         from 
--             jms_dim.dim_tab_pack_money_config_base
--     ) money_config on 1 = 1 and money_config.rn =1 
distribute by 1
;